package com.wenx.activeMqCamel;

import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.ReadContext;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.DefaultMessage;
import org.apache.camel.model.RouteDefinition;

import javax.jms.ConnectionFactory;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;

import static org.apache.activemq.camel.component.ActiveMQComponent.activeMQComponent;

public class activeMQProvider4 {
    private static String user = "admin";
    private static String password = "admin";
    //填写AMQ的IP
    private static String url = "tcp://10.10.3.186:61616?" +
            "wireFormat.maxInactivityDurationInitalDelay=30000";
    public static void main(String[] args) throws Exception {
        CamelContext context = new DefaultCamelContext(); // 1. 创建 CamelContext.
        //创建AMQ的连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
        ActiveMQComponent activeMQComponent = new ActiveMQComponent();
        activeMQComponent.setConnectionFactory(connectionFactory);

        //使用连接池连接
        activeMQComponent.setUsePooledConnection(true);
        //事务
        activeMQComponent.setTransacted(true);
        //客户端确认
        activeMQComponent.setAcknowledgementMode(3);
        //持久化
        activeMQComponent.setDeliveryPersistent(true);
        context.addComponent("activemq", activeMQComponent);
        System.out.println(url + " " + user + " " + password);
        context.setTracing(true);
        context.start(); // 4. 启动CamelContext.
        context.addRouteDefinition(initRoute());
        Thread.sleep(Integer.MAX_VALUE);  // 为了保持CamelContext处于工作状态，这里需要sleep主线程
        context.stop(); // 最后停止CamelContext
    }
    private static int count = 1;
    public static RouteDefinition initRoute(){
        RouteDefinition routeDefinition = new RouteDefinition();
        routeDefinition
                .from("quartz://foo?cron=*/10 * * * * ?&stateful=true")
                //.from("timer:foo?repeatCount=1")
                .doTry()
                .to("http://10.10.3.162:8081/api/restful/getAllUser")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        //解析exchange中的返回值
                        DefaultMessage message = (DefaultMessage) exchange.getIn();
                        InputStream bodyStream =  (InputStream)message.getBody();
                        ByteArrayOutputStream result = new ByteArrayOutputStream();
                        byte[] buffer = new byte[1024];
                        int length;
                        while ((length = bodyStream.read(buffer)) != -1) {
                            result.write(buffer, 0, length);
                        }
                        //rest响应数据集
                        String bodystr = result.toString("UTF-8");
                        //String data = "{\"data\":" + bodystr + "}";
                        result.close();
                        bodyStream.close();
                        System.out.println("第"+count+"次生产的body="+bodystr);
                        count++;
                        ReadContext context = JsonPath.parse(bodystr);
//                        Object mapList =
//                                context.read("$.data[?((@.id > 50 && @.name == '王五') || @.id == 123)]");
                        exchange.getOut().setBody(bodystr);
                    }
                })
                .to("activemq:topic:VirtualTopic.test1")
                .doCatch(Exception.class)
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        Object camelExceptionCaught = exchange.getProperty("CamelExceptionCaught");
                        System.out.println("camelExceptionCaught = " + camelExceptionCaught);
                    }
                })
                .endDoTry();
        return routeDefinition;
    }
}
